package com.rrd.hive.udaf;


import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.serde2.objectinspector.*;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.io.Text;

import java.util.HashMap;
import java.util.Map;

/**
 * @author cqliu
 * @version 创建时间：2019年10月09日
 *
 *
 */
public class NewestValueUDAF extends AbstractGenericUDAFResolver {

    static class UDAFValueState extends GenericUDAFEvaluator.AbstractAggregationBuffer {
        Map<String,String> rMap = new HashMap<String,String>();
    }
    @Override
    public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)
            throws SemanticException {
        if (parameters.length != 2) {
            throw new UDFArgumentTypeException(parameters.length - 1,
                    "Exactly two argument is expected.");
        }

        if (parameters[1].getCategory() != ObjectInspector.Category.PRIMITIVE) {
            throw new UDFArgumentTypeException(0,
                    "Only primitive type arguments are accepted but "
                            + parameters[0].getTypeName() + " is passed.");
        }
        switch (((PrimitiveTypeInfo) parameters[1]).getPrimitiveCategory()) {
            case LONG:
                return new NewestValueUDAF.GetNewestValue();
            case TIMESTAMP:
            case INT:
            case FLOAT:
            case DECIMAL:
            case SHORT:
            case DOUBLE:
            case STRING:
            case VARCHAR:
            case CHAR:
            case BOOLEAN:
            case DATE:
            default:
                throw new UDFArgumentTypeException(0,
                        "Only numeric type arguments are accepted but "
                                + parameters[1].getTypeName() + " is passed.");
        }
    }


    public static class GetNewestValue extends GenericUDAFEvaluator{

        private PrimitiveObjectInspector inputOI1;
        private PrimitiveObjectInspector inputOI2;
        private StandardMapObjectInspector mapOI;
        UDAFValueState state = null;

        @Override
        public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {
            super.init(m, parameters);
            state = new UDAFValueState();
            if (m == Mode.PARTIAL1) {
                this.inputOI1 = (PrimitiveObjectInspector) parameters[0];
                this.inputOI2 = (PrimitiveObjectInspector) parameters[1];

                return ObjectInspectorFactory.getStandardMapObjectInspector(
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector);

            } else if (m == Mode.PARTIAL2) {
                this.mapOI = (StandardMapObjectInspector) parameters[0];
                return ObjectInspectorFactory.getStandardMapObjectInspector(
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            } else if (m == Mode.FINAL) {
                this.mapOI = (StandardMapObjectInspector) parameters[0];
                return ObjectInspectorFactory.getStandardMapObjectInspector(
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            } else if (m == Mode.COMPLETE) {
                return ObjectInspectorFactory.getStandardMapObjectInspector(
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            } else { //
                throw new RuntimeException("no such mode Exception");
            }
        }

        @Override
        public AggregationBuffer getNewAggregationBuffer() throws HiveException {
            UDAFValueState agg = new UDAFValueState();
            reset(agg);
            return agg;
        }

        @Override
        public void reset(AggregationBuffer agg) throws HiveException {
            UDAFValueState bdagg = (UDAFValueState)agg;
            bdagg.rMap.put("T","0");
            bdagg.rMap.put("V","");

        }

        @Override
        public void iterate(AggregationBuffer agg, Object[] objects) throws HiveException {
            UDAFValueState myagg = (UDAFValueState)agg;
            Long otimestamp = myagg.rMap.get("T") == null ? 0 : Long.parseLong(myagg.rMap.get("T"));
            if(objects[0] == null || objects[1] == null){
                return ;
            }
            Long mytimestamp = PrimitiveObjectInspectorUtils.getLong(objects[1],inputOI2);
            String myvalue = PrimitiveObjectInspectorUtils.getString(objects[0],inputOI1);

            if(myvalue != null && !"".equals(myvalue.trim()) && mytimestamp >= otimestamp){
                myagg.rMap.put("T",mytimestamp.toString());
                myagg.rMap.put("V",myvalue.trim());
            }
        }

        @Override
        public Object terminatePartial(AggregationBuffer agg) throws HiveException {
            UDAFValueState myagg = (UDAFValueState)agg;
            return myagg.rMap;
        }

        @Override
        public void merge(AggregationBuffer agg, Object o) throws HiveException {
            UDAFValueState myagg = (UDAFValueState)agg;
            Map<Text,Text> oMap = (Map<Text,Text>)mapOI.getMap(o);
            if(o == null ){
                return ;
            }
            Long otimeStamp = Long.parseLong(oMap.get(new Text("T")).toString());
            String oValue = oMap.get(new Text("V")).toString();
            if(otimeStamp >= Long.parseLong(myagg.rMap.get("T")) && !"".equals(oValue)){
                myagg.rMap.put("T",otimeStamp.toString());
                myagg.rMap.put("V",oValue.trim());
            }
        }

        @Override
        public Object terminate(AggregationBuffer agg) throws HiveException {
            UDAFValueState myagg = (UDAFValueState)agg;
            return myagg.rMap;
        }
    }
}